#include "EventLoopMgr.h"
#include <stdint.h>
#ifdef _WIN32
#include <winsock.h>
#pragma comment(lib,"ws2_32.lib")
#endif
EventLoopMngr EventLoopMngr::m_oInstance;

 class Event
{
public:
	int64_t Wait()
	{
		std::unique_lock<decltype(m_MutexCondition)> lock{ m_MutexCondition };
		m_Condition.wait(lock, [&]()->bool { return ret != -1; }); // suspend and wait ...	
		return ret;
	}
	void Set(int64_t re)
	{
		std::unique_lock<decltype(m_MutexCondition)> lock(m_MutexCondition);
		ret = re;
		m_Condition.notify_one();
	}
	std::mutex m_MutexCondition;
	std::condition_variable m_Condition;
	int64_t ret = -1;
};
TaskScheduler::TaskScheduler()
{
}
TaskScheduler::~TaskScheduler()
{
	StopRun();
}

int64_t TaskScheduler::getCurrentTicket()
{

#ifdef _WIN32

	LARGE_INTEGER nCurrentTime;
	LARGE_INTEGER m_nFreq;
	QueryPerformanceFrequency(&m_nFreq);
	QueryPerformanceCounter(&nCurrentTime);

	uint64_t n_timer = (uint64_t)((nCurrentTime.QuadPart) * 1000 / (double)m_nFreq.QuadPart + 0.5f);

	return n_timer;
#else

	struct ? timeval ? ? tv;
	gettimeofday(&tv, NULL);
	uint64_t timerMs = tv.tv_sec * 1000.0 + tv.tv_usec / 1000.0;
	return timerMs;
#endif

	//std::chrono::steady_clock::time_point t1 = std::chrono::steady_clock::now();
	//std::chrono::steady_clock::time_point t11;
	//int64_t duration = std::chrono::duration_cast<std::chrono::microseconds>(t1 - t11).count();
	//int64_t duration = std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t11).count();
	return  std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
}
int32_t TaskScheduler::StartRun()
{
	m_loopThread = new std::thread(&TaskScheduler::RunLoop, this);
	//std::unique_lock<std::mutex> lock(_mutex);
	//_condition.wait(lock, [this]() { return m_bStart == true; });
	while (!m_bStart)
	{
		std::this_thread::sleep_for(std::chrono::milliseconds(10));
	}

	if (m_loopThread)
	{
		return 0;
	}
	return -1;

}
void TaskScheduler::StopRun()
{
	m_bExitThread = true;
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([]() {int a; });
	}
	_condition.notify_one();

	if (m_loopThread)
	{
		m_loopThread->join();
		delete m_loopThread;
		m_loopThread = 0;
	}
}

int64_t TaskScheduler::AddTimer(int64_t iDelay, std::function<bool()> fcb)
{

	Event tEvent;
	auto t_timer = std::make_shared<Timer>(iDelay, fcb);
	int64_t t_timer_id = 0;
	InvokTask([this, &t_timer, &t_timer_id]()
		{
			m_iTimerId++;
			t_timer->m_iTimerId = m_iTimerId;
			t_timer_id = m_iTimerId;
			uint64_t time_line = getCurrentTicket() + t_timer->m_iDelay;
			m_delayTask.emplace(time_line, std::move(t_timer));
		}
	);
	return t_timer_id;
}

void TaskScheduler::PostTask(Task task)
{
	// 调用线程和将要打入的线程是同一个
	if (_loopThreadId == std::this_thread::get_id())
	{
		// 原地执行就可以
		task();
	}
	{
		std::unique_lock<std::mutex> lock(_mutex);
		Event tEvent;
		_queue.push_back([task]() {
			task();

			});
	}
	_condition.notify_one();
}

void TaskScheduler::InvokTask(Task task)
{
	// 调用线程和将要打入的线程是同一个
	if (_loopThreadId == std::this_thread::get_id())
	{
		// 原地执行就可以
		task();
	}

	Event tEvent;
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([task, &tEvent]() {
			task();
			tEvent.Set(0);
			});
	}
	_condition.notify_one();
	printf("Wait before \n");
	tEvent.Wait();
}

int32_t TaskScheduler::DelTimer(int64_t timerId)
{

	int ret = -1;
	InvokTask([this, timerId, &ret]()
		{
			std::multimap<uint64_t, std::shared_ptr<Timer> >::iterator it = m_delayTask.begin();
			for (; it != m_delayTask.end();)
			{
				std::shared_ptr<Timer> pTimer = it->second;
				if (pTimer->m_iTimerId == timerId)
				{
					it = m_delayTask.erase(it);
					ret = 0;
					return;
				}
				else
				{
					it++;
				}
			}

		});
	return ret;
}


void TaskScheduler::RunLoop()
{
	_loopThreadId = std::this_thread::get_id();
	m_bStart = true;
	//_condition.notify_one();

	while (!m_bExitThread)
	{
		int64_t delay1 = getMinDelay(); // 处理定时器任务
	

		int64_t t_delay = 3333 + 100;
		std::list<Task> t_queue;
		{
			std::unique_lock<std::mutex> lock(_mutex);
			int test1 = 1090;
			//printf("delay:%d test1:%d t_delay:%d this:%p \n", delay1, t_delay, test1,  this);
			//_condition.wait_for(lock, std::chrono::milliseconds(delay), [this]() { return !_queue.empty(); });
			_condition.wait_for(lock, std::chrono::milliseconds(delay1), [this]() { return _queue.size() > 0; });

			printf("_queue.size:%d \n", _queue.size());
			if (_queue.size() > 0)
			{
				printf("1111 _queue.size:%d \n", _queue.size());
				t_queue.swap(_queue);

			}
		}
		// 处理任务
		if (t_queue.size() > 0)
		{
			for (std::list<Task>::iterator it = t_queue.begin(); it != t_queue.end(); it++)
			{
				Task task = *it;
				task();
			}
			t_queue.clear();
		}
	}

}


int32_t TaskScheduler::addTimerInner(std::shared_ptr<Timer>& _Timer)
{
	printf("delay:%lld \n", _Timer->m_iDelay);
	m_iTimerId++;
	_Timer->m_iTimerId = m_iTimerId;
	uint64_t time_line = getCurrentTicket() + _Timer->m_iDelay;
	m_delayTask.emplace(time_line, std::move(_Timer));

	return m_iTimerId;

}


int32_t TaskScheduler::delTimerInner(int64_t timerId)
{
	std::multimap<uint64_t, std::shared_ptr<Timer> >::iterator it = m_delayTask.begin();

	for (; it != m_delayTask.end();)
	{
		std::shared_ptr<Timer> pTimer = it->second;
		if (pTimer->m_iTimerId == timerId)
		{
			it = m_delayTask.erase(it);
			return 0;
		}
		else
		{
			it++;
		}
	}
	return -1;
}

int64_t TaskScheduler::getMinDelay() 
{

	auto it = m_delayTask.begin();
	if (it == m_delayTask.end()) {
		return 0;
	}
	uint64_t now = getCurrentTicket();
	if (it->first > now) {
		//所有任务尚未到期
		printf("it->first:%llu now:%llu dif:%lld \n", it->first, now, it->first - now);
		return it->first - now;
	}
	printf("flushDelayTask before \n");
	//执行已到期的任务并刷新休眠延时
	int64_t ret = this->flushDelayTask(now);
	printf("flushDelayTask after \n");
	return ret;
}

int64_t TaskScheduler::flushDelayTask(uint64_t now_time) {

	printf("############# sleep :%llu \n", now_time);
	decltype(m_delayTask) taskCopy;
	taskCopy.swap(m_delayTask);
	uint64_t start = getCurrentTicket();
	for (auto it = taskCopy.begin(); it != taskCopy.end() && it->first <= now_time; it = taskCopy.erase(it)) {
		//已到期的任务
		try {
			//auto next_delay = (*(it->second))();
			std::shared_ptr<Timer>  pTimer = it->second;
			if (pTimer->m_cb)
			{
				bool ret = pTimer->m_cb();
				if (ret)
				{
					m_delayTask.emplace(pTimer->m_iDelay + now_time, std::move(it->second));
				}
			}
		}
		catch (std::exception& ex) {
			//ErrorL << "EventPoller执行延时任务捕获到异常:" << ex.what();
		}
	}
	uint64_t end = getCurrentTicket();
	int64_t useTime = end - start;
	if (useTime > 11)
	{
		printf("timers_poll::flushDelayTask useTimer = %d \n", useTime);
	}

	taskCopy.insert(m_delayTask.begin(), m_delayTask.end());
	taskCopy.swap(m_delayTask);

	auto it = m_delayTask.begin();
	if (it == m_delayTask.end()) {
		//没有剩余的定时器了
		return 0;
	}
	//最近一个定时器的执行延时

	int64_t t_sleep = it->first - getCurrentTicket();
	if (t_sleep < -1)
	{
		printf("timers_poll::flushDelayTask  t_sleep = %lld  \n", t_sleep);
	}
	printf("t_sleep :%lld \n", t_sleep);
	return t_sleep > 0 ? t_sleep : 0;

}


EventLoop::EventLoop()
{
	num_threads_ = 1;

}

EventLoop::~EventLoop()
{
	this->Quit();
}

std::shared_ptr<TaskScheduler> EventLoop::GetTaskScheduler()
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() == 1) {
		return task_schedulers_.at(0);
	}
	else {
		auto task_scheduler = task_schedulers_.at(index_);
		index_++;
		if (index_ >= task_schedulers_.size()) {
			index_ = 0;
		}
		return task_scheduler;
	}

	return nullptr;
}

int64_t EventLoop::AddTimer(int64_t iDelayUs, std::function<bool()> fcb)
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() > 0) {
		return task_schedulers_[0]->AddTimer(iDelayUs, fcb);
	}
	return -1;
}

int32_t EventLoop::DelTimer(int64_t timerId)
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() > 0) {
		return task_schedulers_[0]->DelTimer(timerId);
	}
	return -1;
}

void EventLoop::PostTask(Task task)
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() > 0) {
		task_schedulers_[0]->PostTask(task);
	}

}

void EventLoop::InvokTask(Task task)
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() > 0) {
		return task_schedulers_[0]->InvokTask(task);
	}
	return;
}

int32_t EventLoop::Loop(uint32_t num_threads)
{
	std::lock_guard<std::mutex> locker(mutex_);
	num_threads_ = num_threads;
	if (!task_schedulers_.empty()) {
		printf("EventLoop Have Looped \n");
		return -1;
	}
	
	for (uint32_t n = 0; n < num_threads_; n++)
	{
		int32_t ret = 0;
		std::shared_ptr<TaskScheduler> task_scheduler_ptr(new TaskScheduler());
		ret = task_scheduler_ptr->StartRun();
		if (0 == ret)
		{
			task_schedulers_.push_back(task_scheduler_ptr);
		}
	}

	return task_schedulers_.size();

}

void EventLoop::Quit()
{
	std::lock_guard<std::mutex> locker(mutex_);

	for (auto iter : task_schedulers_) {
		iter->StopRun();
	}
	task_schedulers_.clear();

}
EventLoopMngr* EventLoopMngr::GetInstance()
{
	return &(EventLoopMngr::m_oInstance);
}

EventLoopMngr::EventLoopMngr()
{

}
EventLoopMngr::~EventLoopMngr()
{
	m_oEventLoop.Quit();
}
int32_t EventLoopMngr::Init(uint32_t num_threads)
{
	int32_t ret = m_oEventLoop.Loop(num_threads);
	return ret;
}
